/*     */ package com.zimbra.cs.redolog.util;
/*     */ 
/*     */ import com.zimbra.common.service.ServiceException;
/*     */ import com.zimbra.common.util.Log;
/*     */ import com.zimbra.common.util.ZimbraLog;
/*     */ import com.zimbra.cs.redolog.RedoPlayer;
/*     */ import com.zimbra.cs.redolog.op.RedoableOp;
/*     */ import com.zimbra.cs.util.Zimbra;
/*     */ import java.util.concurrent.BlockingQueue;
/*     */ import java.util.concurrent.LinkedBlockingQueue;
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ public class ParallelRedoPlayer
/*     */   extends RedoPlayer
/*     */ {
/*     */   private PlayerThread[] mPlayerThreads;
/*     */   
/*     */   public ParallelRedoPlayer(boolean writable, boolean unloggedReplay, boolean ignoreReplayErrors, boolean skipDeleteOps, int numThreads, int queueCapacity, boolean handleMailboxConflict)
/*     */   {
/*  36 */     super(writable, unloggedReplay, ignoreReplayErrors, skipDeleteOps, handleMailboxConflict);
/*  37 */     ZimbraLog.redolog.debug("Starting ParallelRedoPlayer");
/*  38 */     numThreads = Math.max(numThreads, 1);
/*  39 */     this.mPlayerThreads = new PlayerThread[numThreads];
/*  40 */     for (int i = 0; i < numThreads; i++) {
/*  41 */       String name = "RedoPlayer-" + Integer.toString(i);
/*  42 */       PlayerThread player = new PlayerThread(queueCapacity, null);
/*  43 */       this.mPlayerThreads[i] = player;
/*  44 */       player.setName(name);
/*  45 */       player.start();
/*     */     }
/*     */   }
/*     */   
/*     */   public void shutdown() {
/*  50 */     ZimbraLog.redolog.debug("Shutting down ParallelRedoPlayer");
/*     */     try {
/*  52 */       super.shutdown();
/*     */     } finally { int i;
/*  54 */       for (int i = 0; i < this.mPlayerThreads.length; i++) {
/*  55 */         this.mPlayerThreads[i].shutdown();
/*     */       }
/*     */     }
/*  58 */     ZimbraLog.redolog.debug("ParallelRedoPlayer shutdown complete");
/*     */   }
/*     */   
/*     */   protected void playOp(RedoableOp op) throws Exception {
/*  62 */     checkError();
/*  63 */     int mboxId = op.getMailboxId();
/*  64 */     if ((mboxId == -1) || (mboxId == 0))
/*     */     {
/*     */ 
/*  67 */       if (ZimbraLog.redolog.isDebugEnabled())
/*  68 */         ZimbraLog.redolog.info("Executing: " + op.toString());
/*  69 */       op.redo();
/*     */ 
/*     */     }
/*     */     else
/*     */     {
/*     */ 
/*  75 */       int index = Math.abs(mboxId % this.mPlayerThreads.length);
/*  76 */       PlayerThread player = this.mPlayerThreads[index];
/*  77 */       RedoTask task = new RedoTask(op);
/*  78 */       if (ZimbraLog.redolog.isDebugEnabled())
/*  79 */         ZimbraLog.redolog.info("Enqueuing: " + op.toString());
/*     */       try {
/*  81 */         player.enqueue(task);
/*     */       } catch (InterruptedException e) {}
/*     */     }
/*     */   }
/*     */   
/*  86 */   private Throwable mError = null;
/*  87 */   private final Object mErrorLock = new Object();
/*     */   
/*     */   private void raiseError(Throwable t) {
/*  90 */     synchronized (this.mErrorLock) {
/*  91 */       this.mError = t;
/*     */     }
/*     */   }
/*     */   
/*     */   private void checkError() throws ServiceException {
/*  96 */     synchronized (this.mErrorLock) {
/*  97 */       if (this.mError != null) {
/*  98 */         throw ServiceException.FAILURE("Redo playback stopped due to an earlier error: " + this.mError.getMessage(), this.mError);
/*     */       }
/*     */     }
/*     */   }
/*     */   
/*     */   private boolean hadError() {
/* 104 */     synchronized (this.mErrorLock) {
/* 105 */       return this.mError != null;
/*     */     }
/*     */   }
/*     */   
/*     */   private static class RedoTask { private RedoableOp mOp;
/*     */     
/* 111 */     public RedoTask(RedoableOp op) { this.mOp = op; }
/* 112 */     public RedoableOp getOp() { return this.mOp; }
/* 113 */     public boolean isShutdownTask() { return false; }
/*     */   }
/*     */   
/*     */ 
/*     */   private static class ShutdownTask
/*     */     extends ParallelRedoPlayer.RedoTask
/*     */   {
/* 120 */     public ShutdownTask() { super(); }
/*     */     
/* 122 */     public boolean isShutdownTask() { return true; }
/*     */   }
/*     */   
/*     */   private class PlayerThread extends Thread {
/*     */     private BlockingQueue<ParallelRedoPlayer.RedoTask> mQueue;
/*     */     
/*     */     private PlayerThread(int queueCapacity) {
/* 129 */       queueCapacity = Math.max(queueCapacity, 1);
/* 130 */       this.mQueue = new LinkedBlockingQueue(queueCapacity);
/*     */     }
/*     */     
/*     */     public void enqueue(ParallelRedoPlayer.RedoTask task) throws InterruptedException {
/* 134 */       this.mQueue.put(task);
/*     */     }
/*     */     
/*     */     public void shutdown() {
/* 138 */       if (ParallelRedoPlayer.this.hadError())
/* 139 */         this.mQueue.clear();
/* 140 */       ParallelRedoPlayer.ShutdownTask t = new ParallelRedoPlayer.ShutdownTask();
/*     */       try {
/* 142 */         this.mQueue.put(t);
/*     */       } catch (InterruptedException e) {}
/*     */       try {
/* 145 */         join();
/*     */       } catch (InterruptedException e) {}
/*     */     }
/*     */     
/*     */     public void run() {
/*     */       for (;;) {
/*     */         ParallelRedoPlayer.RedoTask task;
/*     */         try {
/* 153 */           task = (ParallelRedoPlayer.RedoTask)this.mQueue.take();
/*     */         } catch (InterruptedException e) {
/*     */           break;
/*     */         }
/* 157 */         if (task.isShutdownTask()) {
/*     */           break;
/*     */         }
/* 160 */         if (!ParallelRedoPlayer.this.hadError())
/*     */         {
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/*     */ 
/* 168 */           RedoableOp op = task.getOp();
/*     */           try {
/* 170 */             if (ZimbraLog.redolog.isDebugEnabled()) {
/* 171 */               ZimbraLog.redolog.info("Executing: " + op.toString());
/*     */             }
/* 173 */             if (ParallelRedoPlayer.this.handleMailboxConflict) {
/* 174 */               ParallelRedoPlayer.this.redoOpWithMboxConflict(op);
/*     */             } else {
/* 176 */               op.redo();
/*     */             }
/*     */           } catch (OutOfMemoryError oome) {
/* 179 */             Zimbra.halt("Out of memory while executing redo op", oome);
/*     */           } catch (Throwable e) {
/* 181 */             ZimbraLog.redolog.error("Unable to execute redo op: " + op.toString(), e);
/* 182 */             if (!ParallelRedoPlayer.this.ignoreReplayErrors()) {
/* 183 */               ParallelRedoPlayer.this.raiseError(e);
/*     */             }
/*     */           }
/*     */         }
/*     */       }
/*     */     }
/*     */   }
/*     */ }


/* Location:              /home/mint/zimbrastore.jar!/com/zimbra/cs/redolog/util/ParallelRedoPlayer.class
 * Java compiler version: 7 (51.0)
 * JD-Core Version:       0.7.1
 */